package cn.edu.laud.sany.dataimport;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.SQLTransientConnectionException;
import java.sql.Statement;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.zip.GZIPInputStream;

import org.apache.cassandra.thrift.TimedOutException;

import cn.edu.laud.sany.types.DataFormatErrorException;
import cn.edu.laud.sany.types.ImportProcess;
import cn.edu.laud.sany.types.RowData;
import cn.edu.laud.sany.types.TimeSerial;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;
import cn.edu.laud.sany.utils.TimeUtils;

public class LaUDRunner extends Thread
{
	public static int PROCESS_LOG_BATCH_SIZE = 10;
	public static int READ_BATCH_SIZE = 1000;
	public static int JDBC_BATCH_SIZE = 10000;
	protected Statement laudStatement;
	private Connection laudConnection;
	List<String> sqls = new ArrayList<String>();
	
	protected MyLog logger;
	protected SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	protected SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
	// 当前处理的时间，用来做标志以确定如何添加流水号
	protected Map<String, TimeSerial> currenttime = new HashMap<String, TimeSerial>();
	// 流水号
	protected TimeSerial timeSerial = null;
	
	protected int threadNum;
	protected String keyspace;
	protected String nowFileName = "";
	protected long nowLineNum = 0;
	
	// protected boolean isImportSucceed = true;
	
	public LaUDRunner(int no, String keyspace)
	{
		this.threadNum = no;
		this.keyspace = keyspace;
		File foler = new File("log");
		if (foler.exists() == false)
		{
			foler.mkdir();
		}
		logger = MyLog.getMyLog(LaUDImport.getLogFilePath(), true);
		logger.setThreadName("Thread " + no);
	}
	
	public boolean openConnection()
	{
		try
		{
			String jdbcString = LaUDImport.host[threadNum % LaUDImport.host.length];
			try
			{
				logger.info("try get jdbc connection...");
				laudConnection = DriverManager.getConnection(jdbcString, "", "");
				laudStatement = laudConnection.createStatement();
				logger.info("get jdbc connection successful.");
			}
			catch (SQLException e1)
			{
				e1.printStackTrace();
				logger.error("get jdbc connection error:" + e1.getMessage() + ", jdbc url:"
						+ jdbcString);
				return false;
			}
			laudStatement.execute("use " + keyspace);
		}
		catch (SQLException e1)
		{
			e1.printStackTrace();
			logger.error(e1.getMessage());
			return false;
		}
		return true;
	}
	
	public void closeConnection()
	{
		try
		{
			if (laudStatement != null)
			{
				try
				{
					laudStatement.close();
				}
				catch (SQLException e)
				{
					e.printStackTrace();
					logger.error(e.getMessage());
				}
			}
			if (laudConnection != null)
			{
				try
				{
					laudConnection.close();
				}
				catch (SQLException e)
				{
					e.printStackTrace();
					logger.error(e.getMessage());
				}
			}
			logger.info("laud connect closed successfully.");
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	
	public void endImport() throws IOException
	{
		int sqlNum = sqls.size();
		if (sqlNum > 0)
		{
			boolean importSucceed = executeSQL(sqls);
			sqls.clear();
			if (importSucceed == false)
			{
				nowLineNum = nowLineNum - sqlNum;
			}
		}
	}
	
	/**
	 * 
	 * @param paraId
	 * @param equipCode
	 * @param paravalue
	 * @param localTime
	 * @return 0 if successful or the error data num
	 * @throws IOException
	 */
	public void addRowData(RowData rowData) throws IOException
	{
		sqls.add(String.format(LaUDImport.cQLInsertFormat, rowData.paraId, rowData.localTime,
				rowData.equipCode, rowData.paravalue));
		int sqlNum = sqls.size();
		if (sqlNum == JDBC_BATCH_SIZE)
		{
			while (executeSQL(sqls) == false)
			{
				logger.info("execute [" + sqlNum + "] sqls error, will try again.");
				try
				{
					Thread.sleep(100);
				}
				catch (InterruptedException e)
				{
				}
			}
			ProcessManager.addRealImportCount(sqlNum);
			sqls.clear();
		}
	}
	
	protected boolean executeSQL(List<String> sqls)
	{
		int timeouttime = 5;
		while (timeouttime > 0)
		{
			StringBuilder sb = new StringBuilder(2000);
			sb.append("BEGIN BATCH USING CONSISTENCY ");
			sb.append(LaUDImport.cQLTranscation);
			sb.append(" ");
			for (String sql : sqls)
			{
				sb.append(sql);
				sb.append(" ");
			}
			sb.append(" APPLY BATCH");
			try
			{
				laudStatement.execute(sb.toString());
				sqls.clear();
				return true;
			}
			catch (SQLException e)
			{
				logger.equals(e.getMessage());
				if (timeouttime > 0 && (e.getCause() instanceof TimedOutException))
				{
					logger.warn("time out exception, will try again after 5 secondes,[time "
							+ timeouttime + "]");
					timeouttime--;
					try
					{
						Thread.sleep(5000);
					}
					catch (InterruptedException e1)
					{
						logger.equals(e.getMessage());
					}
				}
				else if (timeouttime > 0
						&& (e.getCause() instanceof SQLTransientConnectionException))
				{
					logger.warn(e.getMessage() + ", will try again after 5 secondes,[time "
							+ timeouttime + "]");
					timeouttime--;
					try
					{
						Thread.sleep(5000);
					}
					catch (InterruptedException e1)
					{
						logger.equals(e.getMessage());
					}
				}
				else
				{
					logger.error(e.getMessage() + "(sql:" + sb.toString().substring(0, 100)
							+ "...)");
					return false;
				}
			}
		}
		return false;
	}
	
	private long getExistingCount(String fileName, BufferedReader bReader) throws IOException
	{
		// 表头
		bReader.readLine();
		long begin = 0;
		ImportProcess process = ProcessManager.getProcessCount(fileName);
		if (process.status == 1)
		{
			begin = process.count;
			for (int k = 0; k < begin; k++)
			{
				bReader.readLine();
			}
		}
		return begin;
	}
	
	private String trimData(String oriData)
	{
		oriData = oriData.trim();
		oriData = oriData.substring(1, oriData.length() - 1);
		return oriData;
	}
	
	protected RowData getRowData(String lineString) throws DataFormatErrorException
	{
		String data[] = null;
		String equipCode = null;
		int paraId = 0;
		String paravalue = null;
		long localTime = 0L;
		try
		{
			data = lineString.split(",");
			if (data.length != 6)
			{
				ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
				return null;
			}
			equipCode = trimData(data[1]);
			paraId = Integer.valueOf(trimData(data[2]));
			if (LaUDImport.ignoreSomeParaId && LaUDImport.ignoreParaIds.contains(paraId))
			{
				return null;
			}
			paravalue = trimData(data[3]);
			String dataString = trimData(data[5]);
			try
			{
				localTime = sdf.parse(dataString).getTime();
			}
			catch (ParseException e)
			{
				try
				{
					localTime = sdf2.parse(dataString).getTime();
				}
				catch (ParseException e1)
				{
					ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
					throw new DataFormatErrorException();
				}
			}
		}
		catch (NumberFormatException e1)
		{
			ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
			throw new DataFormatErrorException();
		}
		// 添加流水号
		timeSerial = currenttime.get(equipCode + paraId);
		if (timeSerial == null)
		{
			timeSerial = new TimeSerial(localTime, 0);
			currenttime.put(equipCode + paraId, timeSerial);
		}
		else
		{
			if (localTime != timeSerial.time)
			{
				timeSerial.time = localTime;
				timeSerial.serial = 0;
			}
			else
			{
				timeSerial.serial++;
				localTime += timeSerial.serial;
				if (timeSerial.serial >= 1000)
				{
					ProcessManager.dataTooManyLog(nowFileName, nowLineNum, lineString);
				}
			}
		}
		RowData rowData = new RowData();
		rowData.paraId = paraId;
		rowData.paravalue = paravalue;
		rowData.equipCode = equipCode;
		rowData.localTime = localTime;
		return rowData;
	}
	
	protected boolean importFileData(File file)
	{
		nowLineNum = 0;
		String fileName = file.getAbsolutePath();
		nowFileName = fileName;
		GZIPInputStream gzin = null;
		BufferedReader bReader = null;
		boolean stopFlag = false;
		try
		{
			try
			{
				if (LaUDImport.csvCompressed)
				{
					gzin = new GZIPInputStream(new FileInputStream(file));
					InputStreamReader isreader = new InputStreamReader(gzin);
					bReader = new BufferedReader(isreader);
				}
				else
				{
					bReader = new BufferedReader(new FileReader(file));
				}
			}
			catch (FileNotFoundException e3)
			{
				logger.error("get BufferedReader failed: " + e3.getMessage());
				return false;
			}
			try
			{
				// 跳过已经导入的
				nowLineNum = getExistingCount(fileName, bReader);
			}
			catch (IOException e2)
			{
				logger.error("get ExistingCount failed: " + e2.getMessage());
				bReader.close();
				return false;
			}
			long time = System.nanoTime();
			try
			{
				List<String> lineDataList = new ArrayList<String>();
				int listDataNum = 0;
				int batchNum = 0;
				boolean dataFlag = true;
				while (dataFlag && !stopFlag)
				{
					String lineData = bReader.readLine();
					if (lineData != null)
					{
						listDataNum++;
						lineDataList.add(lineData);
						if (listDataNum < READ_BATCH_SIZE)
						{
							continue;
						}
					}
					else
					{
						dataFlag = false;
					}
					listDataNum = 0;
					List<RowData> rowDataList = new ArrayList<RowData>();
					for (String line : lineDataList)
					{
						RowData rowData = null;
						try
						{
							rowData = getRowData(line);
						}
						catch (DataFormatErrorException e)
						{
							if (LaUDImport.errorDataStop)
							{
								stopFlag = true;
								break;
							}
						}
						if (rowData != null)
						{
							rowDataList.add(rowData);
						}
						nowLineNum++;
					}
					lineDataList.clear();
					
					for (RowData rowData : rowDataList)
					{
						addRowData(rowData);
					}
					batchNum++;
					if (batchNum == PROCESS_LOG_BATCH_SIZE)
					{
						ProcessManager.updateFileCount(fileName, 1, nowLineNum);
						batchNum = 0;
					}
				}
			}
			catch (IOException e1)
			{
				stopFlag = true;
				logger.error("error when imort file " + fileName + " message : " + e1.getMessage());
			}
			finally
			{
				try
				{
					if (LaUDImport.csvCompressed)
					{
						gzin.close();
					}
					else
					{						
						bReader.close();
					}
					endImport();
				}
				catch (IOException e)
				{
					stopFlag = true;
					e.printStackTrace();
				}
				catch (NullPointerException e1)
				{
					stopFlag = true;
					e1.printStackTrace();
					logger.error(e1.getMessage());
				}
			}
			logger.info(String.format("LaUDRunner end successful. (file,size,cost): (%s,%s,%s)",
					threadNum, file.getAbsolutePath(), String.valueOf(nowLineNum),
					TimeUtils.nanoSecondToString(System.nanoTime() - time)));
		}
		catch (Throwable t)
		{
			stopFlag = true;
			t.printStackTrace();
			logger.error(t.getMessage());
		}
		
		return (stopFlag == false);
	}
	
	@Override
	public void run()
	{
		logger.info(String.format("Thread Start...", threadNum));
		File file = null;
		String fileName = null;
		try
		{
			while (openConnection() == false)
			{
				logger.info("begin import failed, thread will sleep 3 sec to restart...");
				try
				{
					Thread.sleep(3000);
				}
				catch (InterruptedException e)
				{
				}
			}
			
			while (true)
			{
				// get source
				file = null;
				fileName = null;
				logger.info("wait to get available csv source file...");
				file = ProcessManager.getAvailableSource();
				if (file == null)
				{
					try
					{
						Thread.sleep(10000);
					}
					catch (InterruptedException e)
					{
						e.printStackTrace();
					}
					continue;
				}
				fileName = file.getAbsolutePath();
				logger.info("find a new input file: " + fileName);
				
				if (ProcessManager.importStartTime == 0)
				{
					ProcessManager.importStartTime = System.nanoTime();
				}
				// import
				boolean succeed = importFileData(file);
				
				// commit task
				if (succeed)
				{
					ProcessManager.commitTask(fileName, true);
					logger.info(String.format("import file %s finished!", fileName));
				}
				else
				{
					ProcessManager.commitTask(fileName, false);
					logger.info(String.format("import file %s failed!", fileName));
				}
			}
		}
		catch (Exception e)
		{
			if (fileName != null)
			{
				ProcessManager.commitTask(fileName, false);
			}
			logger.error(e.getMessage());
			e.printStackTrace();
		}
		finally
		{
			closeConnection();
		}
	}
}
